package com.feidee.data.report.util;

import com.feidee.data.email.model.EConstant;
import com.feidee.data.email.util.ApplicationContextUtil;
import com.feidee.data.report.constant.Constant;
import com.feidee.data.report.model.ReportConfiguration;
import com.feidee.data.report.util.mail.RMailSender;
import com.feidee.data.report.util.threadutil.ThreadPoolUtil;
import com.feidee.fdfalcon.falcon.FalconAlarmService;
import org.apache.log4j.Logger;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.PriorityBlockingQueue;

/**
 * 报表监控线程
 */
public class MonitorThread implements Runnable {

	private static Logger logger = Logger.getLogger(MonitorThread.class);

	private DataSource confDataSource;
	private String groupName;
	private int threadNum;
	private PriorityBlockingQueue<ArrayList<ReportConfiguration>> hqlSqlQueue;

	public MonitorThread(String groupName, PriorityBlockingQueue<ArrayList<ReportConfiguration>> hqlSqlQueue, int threadNum) {
		this.groupName = groupName;
		this.hqlSqlQueue = hqlSqlQueue;
		this.threadNum = threadNum;
	}

	@Override
	public void run() {
		try {
			boolean remainIsSend = false;
			HashSet<String> alreadySendTime = new HashSet<>();
			//直到一组报表所有的线程全部执行完成
			while (threadNum != ThreadPoolUtil.threadCompletedCnt(this.groupName)) {
				String now = TimeUtil.getNow("HH:mm");
				if (now.equals("07:00") && !remainIsSend) {
					if (threadNum == Constant.THREAD_NUM) {
						Connection conn = null;
						PreparedStatement preparedStatement = null;
						ResultSet resultSet = null;
						try {
							confDataSource = (DataSource) ApplicationContextUtil.getBean("confDataSource");
							conn = confDataSource.getConnection();
							String getRemainReportNum = "SELECT COUNT(DISTINCT report_key) AS remain_num FROM report_exec_status " +
									"WHERE exec_status IN (\"DDZX\",\"ZZZX\") AND SUBSTR(update_time,1,10)=\"" + TimeUtil.getNow("yyyy-MM-dd") + "\"";
							preparedStatement = conn.prepareStatement(getRemainReportNum);
							resultSet = preparedStatement.executeQuery();
							if (resultSet.next()) {
								String remainNum = resultSet.getString("remain_num");
								String content = "截止北京时间 07:00,数据门户未执行的报表个数为: " + remainNum + ",请继续留意!";
								RMailSender.sendHtmlMail("七点未执行报表数量监控", content, EConstant.Exception_REPORT_MAIL_HANDLER_RECEIVER);
								logger.info("#### 七点定时发送预警邮件成功 ####");
							}
							remainIsSend = true;
						} catch (Exception e) {
							logger.info("#### 七点定时发送预警邮件失败 ####", e);
						} finally {
							DbOperationUtil.cleanConnEnv(conn, preparedStatement, resultSet);
						}
					}
				}
				if (now.equals("08:00") || now.equals("09:00") || now.equals("10:00")) {
					if (alreadySendTime.contains(now)) {
						continue;
					}
					Connection conn = null;
					PreparedStatement preparedStatement = null;
					ResultSet resultSet = null;
					try {
						confDataSource = (DataSource) ApplicationContextUtil.getBean("confDataSource");
						conn = confDataSource.getConnection();
						String getTable = "SELECT a.report_key,exec_status,priority\n" +
								"FROM (\n" +
								"SELECT report_key,CASE WHEN exec_status=\"DDZX\" THEN \"等待执行\" WHEN exec_status=\"ZZZX\" THEN \"正在执行\" END " +
								"AS exec_status FROM report_exec_status WHERE exec_status IN (\"DDZX\",\"ZZZX\") AND SUBSTR(update_time,1,10)=\""
								+ TimeUtil.getNow("yyyy-MM-dd") + "\" \n" +
								") a\n" +
								"INNER JOIN (\n" +
								"SELECT DISTINCT report_key,priority FROM report_sql\n" +
								") b\n" +
								"ON a.report_key=b.report_key\nORDER BY priority DESC";
						preparedStatement = conn.prepareStatement(getTable);
						resultSet = preparedStatement.executeQuery();
						List<String> successReport = getSuccessReport();
						RMailSender.sendReportExceptionMail(resultSet,
								ThreadPoolUtil.threadExceptionReport(this.groupName),
								ThreadPoolUtil.threadSuccessReport(this.groupName), successReport, false, false);
						logger.info("#### 定时发送预警邮件成功 ####");
					} catch (Exception e) {
						logger.info("#### 定时发送预警邮件失败 ####", e);
					} finally {
						DbOperationUtil.cleanConnEnv(conn, preparedStatement, resultSet);
					}
					alreadySendTime.add(now);
				}
				Thread.sleep(Constant.MONITOR_THREAD_SLEEP_TIME);
			}

			List<String> successReport = getSuccessReport();
			//发送top50报表预警邮件
			RMailSender.sendReportExceptionMail(null,
					ThreadPoolUtil.threadExceptionReport(this.groupName),
					ThreadPoolUtil.threadSuccessReport(this.groupName), successReport, true, false);
			//发送所有报表预警邮件
			RMailSender.sendReportExceptionMail(null,
					ThreadPoolUtil.threadExceptionReport(this.groupName),
					ThreadPoolUtil.threadSuccessReport(this.groupName), successReport, false, true);
            // 发送spark引擎执行结果条数为0的预警
			if (threadNum == Constant.THREAD_NUM && Constant.MONITOR_SQL.size() > 0) {
				FalconAlarmService.sendAlarm("以下spark引擎执行的SQL运行结果为0条，请关注!" + Constant.MONITOR_SQL, "monitor_sql");
				Constant.MONITOR_SQL.clear();
			}
			logger.info("#### 报表运行全部结束，发送预警邮件成功 ####");
		} catch (Exception e) {
			logger.info("#### 报表运行全部结束，发送预警邮件失败 ####", e);
		} finally {
			//清除已发送报表信息
			ThreadPoolUtil.deleteReportInfoCache(this.groupName);
			logger.info("清除监控报表标志成功!");
		}
	}

	public List<String> getSuccessReport() {
		Connection connection = null;
		PreparedStatement preparedStatement = null;
		ResultSet resultSet = null;
		ArrayList<String> successReport = new ArrayList<>();
		try {
			confDataSource = (DataSource) ApplicationContextUtil.getBean("confDataSource");
			connection = confDataSource.getConnection();
			String getTable = "SELECT report_key FROM report_exec_status WHERE exec_status=\"ZXCG\" AND SUBSTR(update_time,1,10)=\""
					+ TimeUtil.getNow("yyyy-MM-dd") + "\"";
			preparedStatement = connection.prepareStatement(getTable);
			resultSet = preparedStatement.executeQuery();
			while (resultSet.next()) {
				String reportKey = resultSet.getString("report_key");
				successReport.add(reportKey);
			}
		} catch (Exception e) {
			logger.error("查询执行成功报表失败！", e);
		} finally {
			DbOperationUtil.cleanConnEnv(connection, preparedStatement, resultSet);
		}
		return successReport;
	}
}
